package com.talent.study.disruptor.height;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;

import java.util.UUID;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.LockSupport;

/**
 * @Author: hszhou
 * @Date: 2025/4/19 11:44
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Slf4j
public class Main2 {

    public static void main(String[] args) throws Exception {



        //构建ringBuffer
        RingBuffer<Trade> ringBuffer = RingBuffer.create(ProducerType.MULTI,
                Trade::new,1024,new YieldingWaitStrategy()
                );

        //通过RingBuffer创建一个屏障
        SequenceBarrier sequenceBarrier =  ringBuffer.newBarrier();


        //构建多消费者
        int c = 10;
        Consumer[] consumers = new Consumer[c];
        for (int i = 0; i <c; i++) {
            consumers[i] = new Consumer("C"+i);
        }
        //构建多消费工作池

        WorkerPool<Trade> workerPool = new WorkerPool<>(
                ringBuffer,sequenceBarrier,new EventExceptionHandler(),consumers
        );


        //设置多个消费者的sequence序号，用于单独统计消费者进度，并且设置到ringBuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        //消费者启动
        workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));


        int count =2;
        CyclicBarrier cyclicBarrier = new CyclicBarrier(count);

        LockSupport.park();
        for (int i = 0; i < count; i++) {

            Producer producer = new Producer(ringBuffer);

            new Thread(()->{
                try{
                    cyclicBarrier.await();
                }catch (Exception e){
                    log.error("[{}]",e);
                }

                for (int j = 0; j < 100; j++) {
                    producer.sendData(UUID.randomUUID().toString());
                }
            }).start();

        }

        log.info("线程创建完毕");








    }

    @Slf4j
    static class  EventExceptionHandler implements  ExceptionHandler<Trade>{


        @Override
        public void handleEventException(Throwable ex, long sequence, Trade event) {

            //消费的时候出现异常
        }

        @Override
        public void handleOnStartException(Throwable ex) {

            //启动的时候
        }

        @Override
        public void handleOnShutdownException(Throwable ex) {

        }
    }
}
